package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.common.satoken.utils.LoginHelper;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.DataModel;
import com.wande.dataplatform.domain.DataModelVersion;
import com.wande.dataplatform.domain.dto.DataModelDTO;
import com.wande.dataplatform.domain.dto.ModelFieldDTO;
import com.wande.dataplatform.mapper.DataModelMapper;
import com.wande.dataplatform.mapper.DataModelVersionMapper;
import com.wande.dataplatform.service.IDataModelService;
import com.wande.dataplatform.service.IDorisService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 数据模型Service实现
 *
 * @author ruoyi
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataModelServiceImpl implements IDataModelService {

    private final DataModelMapper dataModelMapper;
    private final DataModelVersionMapper versionMapper;
    private final IDorisService dorisService;

    private static final Set<String> VALID_TABLE_TYPES = new HashSet<>(Arrays.asList("DUPLICATE", "AGGREGATE", "UNIQUE"));
    private static final Set<String> VALID_PARTITION_TYPES = new HashSet<>(Arrays.asList("RANGE", "LIST"));

    @Override
    public TableDataInfo<DataModel> queryPageList(DataModel dataModel, PageQuery pageQuery) {
        LambdaQueryWrapper<DataModel> wrapper = buildQueryWrapper(dataModel);
        Page<DataModel> page = dataModelMapper.selectPage(pageQuery.build(), wrapper);
        return TableDataInfo.build(page);
    }

    @Override
    public List<DataModel> queryList(DataModel dataModel) {
        LambdaQueryWrapper<DataModel> wrapper = buildQueryWrapper(dataModel);
        return dataModelMapper.selectList(wrapper);
    }

    @Override
    public DataModel queryById(Long id) {
        DataModel dataModel = dataModelMapper.selectById(id);
        if (dataModel == null) {
            throw new ServiceException(DataPlatformErrorCode.DATA_MODEL_NOT_FOUND.getMessage());
        }
        return dataModel;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long createDataModel(DataModelDTO dto) {
        // 验证模型
        validateModel(dto);

        // 检查表名是否已存在
        Long count = dataModelMapper.selectCount(Wrappers.lambdaQuery(DataModel.class)
            .eq(DataModel::getTableName, dto.getTableName()));
        if (count > 0) {
            throw new ServiceException("表名已存在");
        }

        // 创建数据模型
        DataModel dataModel = new DataModel();
        BeanUtil.copyProperties(dto, dataModel);
        dataModel.setFields(JSONUtil.toJsonStr(dto.getFields()));
        dataModel.setVersion(1);
        dataModel.setSyncStatus(0);
        dataModel.setStatus(1);

        dataModelMapper.insert(dataModel);

        // 保存版本历史
        saveVersion(dataModel.getId(), 1, dto.getFields(), "初始版本");

        // 如果需要同步到Doris
        if (Boolean.TRUE.equals(dto.getSyncToDoris())) {
            try {
                syncToDoris(dataModel.getId());
            } catch (Exception e) {
                log.error("同步到Doris失败", e);
                // 不影响创建流程
            }
        }

        return dataModel.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateDataModel(DataModelDTO dto) {
        if (dto.getId() == null) {
            throw new ServiceException("模型ID不能为空");
        }

        // 验证模型
        validateModel(dto);

        DataModel existModel = queryById(dto.getId());

        // 检查表名是否与其他模型冲突
        Long count = dataModelMapper.selectCount(Wrappers.lambdaQuery(DataModel.class)
            .eq(DataModel::getTableName, dto.getTableName())
            .ne(DataModel::getId, dto.getId()));
        if (count > 0) {
            throw new ServiceException("表名已存在");
        }

        // 更新数据模型
        DataModel dataModel = new DataModel();
        BeanUtil.copyProperties(dto, dataModel);
        dataModel.setFields(JSONUtil.toJsonStr(dto.getFields()));
        
        // 如果字段有变化，增加版本号
        if (!existModel.getFields().equals(dataModel.getFields())) {
            dataModel.setVersion(existModel.getVersion() + 1);
            dataModel.setSyncStatus(0); // 重置同步状态
            
            // 保存版本历史
            saveVersion(dataModel.getId(), dataModel.getVersion(), dto.getFields(), 
                StrUtil.isNotBlank(dto.getChangeDescription()) ? dto.getChangeDescription() : "字段变更");
        }

        int result = dataModelMapper.updateById(dataModel);
        return result > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteById(Long id) {
        DataModel dataModel = queryById(id);
        
        // 如果已同步到Doris，先删除Doris表
        if (dataModel.getSyncStatus() == 1 && StrUtil.isNotBlank(dataModel.getDorisDatabase())) {
            try {
                dorisService.dropTable(dataModel.getDorisDatabase(), dataModel.getDorisTable());
            } catch (Exception e) {
                log.error("删除Doris表失败", e);
            }
        }

        // 删除版本历史
        versionMapper.delete(Wrappers.lambdaQuery(DataModelVersion.class)
            .eq(DataModelVersion::getModelId, id));

        // 删除数据模型
        int result = dataModelMapper.deleteById(id);
        return result > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteByIds(List<Long> ids) {
        if (CollUtil.isEmpty(ids)) {
            return false;
        }
        for (Long id : ids) {
            deleteById(id);
        }
        return true;
    }

    @Override
    public Boolean validateModel(DataModelDTO dto) {
        // 验证模型名称
        if (StrUtil.isBlank(dto.getName())) {
            throw new ServiceException("模型名称不能为空");
        }

        // 验证表名
        if (StrUtil.isBlank(dto.getTableName())) {
            throw new ServiceException("表名不能为空");
        }
        if (!dto.getTableName().matches("^[a-zA-Z][a-zA-Z0-9_]*$")) {
            throw new ServiceException("表名格式不正确，必须以字母开头，只能包含字母、数字和下划线");
        }

        // 验证表类型
        if (StrUtil.isBlank(dto.getTableType()) || !VALID_TABLE_TYPES.contains(dto.getTableType())) {
            throw new ServiceException("表类型无效，必须是DUPLICATE、AGGREGATE或UNIQUE");
        }

        // 验证分区类型
        if (StrUtil.isNotBlank(dto.getPartitionType()) && !VALID_PARTITION_TYPES.contains(dto.getPartitionType())) {
            throw new ServiceException("分区类型无效，必须是RANGE或LIST");
        }

        // 验证字段
        if (CollUtil.isEmpty(dto.getFields())) {
            throw new ServiceException("字段列表不能为空");
        }

        Set<String> fieldNames = new HashSet<>();
        boolean hasPrimaryKey = false;
        for (ModelFieldDTO field : dto.getFields()) {
            // 验证字段名
            if (StrUtil.isBlank(field.getName())) {
                throw new ServiceException("字段名称不能为空");
            }
            if (!field.getName().matches("^[a-zA-Z][a-zA-Z0-9_]*$")) {
                throw new ServiceException("字段名格式不正确: " + field.getName());
            }
            if (fieldNames.contains(field.getName())) {
                throw new ServiceException("字段名重复: " + field.getName());
            }
            fieldNames.add(field.getName());

            // 验证字段类型
            if (StrUtil.isBlank(field.getType())) {
                throw new ServiceException("字段类型不能为空: " + field.getName());
            }

            // 检查主键
            if (Boolean.TRUE.equals(field.getPrimaryKey())) {
                hasPrimaryKey = true;
            }

            // AGGREGATE表必须指定聚合类型
            if ("AGGREGATE".equals(dto.getTableType()) && !Boolean.TRUE.equals(field.getPrimaryKey())) {
                if (StrUtil.isBlank(field.getAggregateType())) {
                    throw new ServiceException("聚合表的非主键字段必须指定聚合类型: " + field.getName());
                }
            }
        }

        // UNIQUE和AGGREGATE表必须有主键
        if (("UNIQUE".equals(dto.getTableType()) || "AGGREGATE".equals(dto.getTableType())) && !hasPrimaryKey) {
            throw new ServiceException(dto.getTableType() + "表必须定义主键");
        }

        // 验证分桶字段
        if (StrUtil.isNotBlank(dto.getBucketColumn()) && !fieldNames.contains(dto.getBucketColumn())) {
            throw new ServiceException("分桶字段不存在: " + dto.getBucketColumn());
        }

        // 验证分区字段
        if (StrUtil.isNotBlank(dto.getPartitionColumn()) && !fieldNames.contains(dto.getPartitionColumn())) {
            throw new ServiceException("分区字段不存在: " + dto.getPartitionColumn());
        }

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean syncToDoris(Long id) {
        DataModel dataModel = queryById(id);

        try {
            // 获取或创建租户数据库
            Long tenantId = Long.parseLong(LoginHelper.getTenantId());
            String dbName = dorisService.getTenantDatabaseName(tenantId);
            if (!dorisService.databaseExists(dbName)) {
                dorisService.createDatabase(dbName);
            }

            // 如果表已存在，先删除
            if (dorisService.tableExists(dbName, dataModel.getTableName())) {
                dorisService.dropTable(dbName, dataModel.getTableName());
            }

            // 创建表
            dorisService.createTable(dbName, dataModel);

            // 更新同步状态
            DataModel updateModel = new DataModel();
            updateModel.setId(id);
            updateModel.setDorisDatabase(dbName);
            updateModel.setDorisTable(dataModel.getTableName());
            updateModel.setSyncStatus(1);
            updateModel.setSyncMessage("同步成功");
            dataModelMapper.updateById(updateModel);

            return true;
        } catch (Exception e) {
            log.error("同步到Doris失败", e);
            
            // 更新同步状态为失败
            DataModel updateModel = new DataModel();
            updateModel.setId(id);
            updateModel.setSyncStatus(2);
            updateModel.setSyncMessage("同步失败: " + e.getMessage());
            dataModelMapper.updateById(updateModel);

            throw new ServiceException("同步到Doris失败: " + e.getMessage());
        }
    }

    @Override
    public List<DataModelVersion> queryVersionHistory(Long modelId) {
        return versionMapper.selectList(Wrappers.lambdaQuery(DataModelVersion.class)
            .eq(DataModelVersion::getModelId, modelId)
            .orderByDesc(DataModelVersion::getVersion));
    }

    @Override
    public Boolean updateStatus(Long id, Integer status) {
        DataModel dataModel = new DataModel();
        dataModel.setId(id);
        dataModel.setStatus(status);
        int result = dataModelMapper.updateById(dataModel);
        return result > 0;
    }

    /**
     * 构建查询条件
     */
    private LambdaQueryWrapper<DataModel> buildQueryWrapper(DataModel dataModel) {
        LambdaQueryWrapper<DataModel> wrapper = Wrappers.lambdaQuery();
        wrapper.like(StrUtil.isNotBlank(dataModel.getName()), DataModel::getName, dataModel.getName())
            .like(StrUtil.isNotBlank(dataModel.getTableName()), DataModel::getTableName, dataModel.getTableName())
            .eq(StrUtil.isNotBlank(dataModel.getTableType()), DataModel::getTableType, dataModel.getTableType())
            .eq(dataModel.getStatus() != null, DataModel::getStatus, dataModel.getStatus())
            .eq(dataModel.getProjectId() != null, DataModel::getProjectId, dataModel.getProjectId())
            .eq(dataModel.getSyncStatus() != null, DataModel::getSyncStatus, dataModel.getSyncStatus())
            .orderByDesc(DataModel::getCreateTime);
        return wrapper;
    }

    /**
     * 保存版本历史
     */
    private void saveVersion(Long modelId, Integer version, List<ModelFieldDTO> fields, String changeDescription) {
        DataModelVersion versionEntity = new DataModelVersion();
        versionEntity.setModelId(modelId);
        versionEntity.setVersion(version);
        versionEntity.setFields(JSONUtil.toJsonStr(fields));
        versionEntity.setChangeDescription(changeDescription);
        versionMapper.insert(versionEntity);
    }

}
